package com.yequan.thread.threadpool.mythreadpoolmanager;

import com.yequan.thread.threadpool.plan.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @author 汪海霖 wanghl15@midea.com.cn
 * @ClassName: CustomThreadPoolManager
 * @Description: 线程池工具类
 * @date 2014-5-24 上午10:17:42
 */
public class CustomThreadPoolManager {

    private static Logger logger = LoggerFactory.getLogger(CustomThreadPoolManager.class);
    /**
     * 线程池执行对象
     **/
    private ThreadPoolExecutor threadPool;
    /**
     *
     **/
    private ArrayBlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<Runnable>(10000);
    /**
     * 核心线程个数
     **/
    private int corePoolSize;
    /**
     * 保持心跳时间
     **/
    private static long keepAliveTime = 1L;
    /**
     * 定时执行线程个数
     **/
    private final static int minSchedule = 2;
    /**
     * 延时执行线程
     **/
    private ScheduledExecutorService appSchedule;

    /**
     * 线程池构造方法
     */
    public CustomThreadPoolManager(int nThreads, int capacity, ThreadFactory threadFactory) {
        this.corePoolSize = nThreads;
        RejectedExecutionHandler myHandler = new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                taskQueue.offer(r);
                System.err.println("有任务被拒绝执行");
            }
        };
        Runnable command = new Runnable() {
            @Override
            public void run() {
                Runnable task = null;
                try {
                    task = taskQueue.take();//使用具备阻塞特性的方法
                } catch (InterruptedException e) {
                    return;
                }
                threadPool.execute(task);
            }
        };
        ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(1);
//        每一次执行终止和下一次执行开始之间都存在给定的延迟 16毫秒
        scheduledPool.scheduleWithFixedDelay(command, 0L, 16L, TimeUnit.MILLISECONDS);
        threadPool = new ThreadPoolExecutor(corePoolSize, nThreads,
                keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(capacity), threadFactory);

        threadPool = new ThreadPoolExecutor(corePoolSize, nThreads,
                keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(capacity), threadFactory, myHandler) {
            @Override
            public void afterExecute(Runnable r, Throwable t) {
                super.afterExecute(r, t);
                printException(r, t);
            }
        };
        appSchedule = Executors.newScheduledThreadPool(minSchedule);
    }

    /**
     * 线程池构造方法
     *
     * @param nThreads
     * @param capacity
     */
    public CustomThreadPoolManager(int nThreads, int capacity) {
        this(nThreads, capacity, new NamedThreadFactory());
    }

    private static void printException(Runnable r, Throwable t) {
        if (t == null && r instanceof Future<?>) {
            try {
                Future<?> future = (Future<?>) r;
                if (future.isDone()) {
                    future.get();
                }
            } catch (CancellationException ce) {
                t = ce;
            } catch (ExecutionException ee) {
                t = ee.getCause();
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }
        if (t != null) {
            Throwable cause = t.getCause();
            if (ThreadPoolUtil.isTimeoutThrowable(cause)) {
                logger.error("系统自有线程池任务调用超时异常,error_msg==" + cause.getMessage());
            } else {
                logger.error("系统自有线程池任务异常,error_msg==" + t.getMessage(), t);
            }
        }
    }

    public int getActiveCount() {
        return threadPool.getActiveCount();
    }

    public int getCorePoolCount() {
        return corePoolSize;
    }

    public long getTaskCount() {
        return threadPool.getQueue().size();
    }

    /**
     * 任务添加到线程池中
     *
     * @param paramRunnable
     */
    public Future<?> addExecuteTask(Runnable paramRunnable) {
        if (paramRunnable == null) {
            return null;
        }
        return this.threadPool.submit(paramRunnable);
    }

    public Future<?> addExecuteTask(Callable<?> paramRunnable) {
        if (paramRunnable == null) {
            return null;
        }
        return this.threadPool.submit(paramRunnable);
    }

    public void executeTask(Runnable runnable) {
        if (null == runnable) {
            throw new NullPointerException("任务为空");
        }
        this.threadPool.execute(runnable);
    }

    /**
     * 延时任务添加到线程池中
     *
     * @param task
     * @param delayTime
     */
    public void addDelayExecuteTask(Runnable task, int delayTime) {
        appSchedule.schedule(new DelayTask(task), delayTime,
                TimeUnit.MILLISECONDS);
    }

    /**
     * 延时固定周期执行
     *
     * @param task
     * @param delay
     * @param period
     */
    public void addPeriodDelayExecuteTask(Runnable task, Long delay, Long period) {
        this.appSchedule.scheduleWithFixedDelay(new DelayTask(task), delay, period, TimeUnit.MILLISECONDS);
    }

    public boolean isAsyncBySimple() {
        return getActiveCount() <= getCorePoolCount();
    }

    public void shutdown() {
        taskQueue.clear();
        threadPool.shutdownNow();
    }

    /**
     * 延时任务
     **/
    class DelayTask implements Runnable {

        private Runnable task;

        public DelayTask(Runnable paramTask) {
            this.task = paramTask;
        }

        @Override
        public void run() {
            addExecuteTask(task);
        }
    }
}
